package com.cloud.job;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.lang.UUID;
import cn.hutool.json.JSONUtil;
import com.cloud.constant.ElasticSearchConstants;
import com.cloud.constant.RedisKeyConstants;
import com.cloud.es.ESMoviceInfo;
import com.cloud.pojo.movie.MovieAttr;
import com.google.common.collect.Lists;
import org.elasticsearch.action.admin.indices.get.GetIndexRequest;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.reindex.DeleteByQueryRequest;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.ListOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * @version v1.0
 * @author: TianXiang
 * @description: 缓存数据同步到搜索引擎
 * @date: 2020/11/30
 */

@Component
public class RedisToElasticSearchJob {

    private static final Logger LOG = LoggerFactory.getLogger(RedisToElasticSearchJob.class.getName());

    @Autowired
    private RedisTemplate<String, String> redisTemplate;

    @Autowired
    private RestHighLevelClient client;

    /**
     * 每天凌晨2点同步
     */
    @Scheduled(cron = "0 0 2 * * ?")
    public void execute() throws Exception {
        LOG.info("定时任务执行: 缓存数据同步到搜索引擎");
        LOG.info("重置索引库");
        DeleteByQueryRequest request = new DeleteByQueryRequest(ElasticSearchConstants.INDEX_LIBRARY_NAME);
        request.setQuery(QueryBuilders.existsQuery("movieName"));
        request.setRefresh(true);
        boolean requestExists = client.indices().exists(new GetIndexRequest().indices(ElasticSearchConstants.INDEX_LIBRARY_NAME), RequestOptions.DEFAULT);
        if (requestExists) {
            client.deleteByQuery(request, RequestOptions.DEFAULT);
        }
        DeleteByQueryRequest greenRequest = new DeleteByQueryRequest(ElasticSearchConstants.GREEN_INDEX_LIBRARY_NAME);
        // 设置查询条件
        greenRequest.setQuery(QueryBuilders.existsQuery("movieName"));
        greenRequest.setRefresh(true);
        boolean greenRequestExists = client.indices().exists(new GetIndexRequest().indices(ElasticSearchConstants.GREEN_INDEX_LIBRARY_NAME), RequestOptions.DEFAULT);
        if (greenRequestExists) {
            client.deleteByQuery(greenRequest, RequestOptions.DEFAULT);
        }
        LOG.info("重置索引库-成功");

        ListOperations<String, String> opsForList = redisTemplate.opsForList();
        List<String> cacheDataList = opsForList.range(RedisKeyConstants.MOVIE_LIST_KEY, 0L, opsForList.size(RedisKeyConstants.MOVIE_LIST_KEY));
        List<String> greenCacheDataList = opsForList.range(RedisKeyConstants.GREEN_MOVIE_LIST_KEY, 0L, opsForList.size(RedisKeyConstants.GREEN_MOVIE_LIST_KEY));
        List<MovieAttr> movieAttrList = Lists.newArrayList();
        List<MovieAttr> greenMovieAttrList = Lists.newArrayList();
        cacheDataList.forEach(item -> movieAttrList.add(JSONUtil.toBean(item, MovieAttr.class)));
        greenCacheDataList.forEach(item -> greenMovieAttrList.add(JSONUtil.toBean(item, MovieAttr.class)));

        // 设置字段值
        List<ESMoviceInfo> esMovieInfoList = Lists.newArrayList();
        List<ESMoviceInfo> esGreenMovieInfoList = Lists.newArrayList();
        movieAttrList.forEach(item -> esMovieInfoList.add(new ESMoviceInfo(item.getMovieName().split(".m3u8")[0], JSONUtil.toJsonStr(item))));
        greenMovieAttrList.forEach(item -> esGreenMovieInfoList.add(new ESMoviceInfo(item.getMovieName().split(".m3u8")[0], JSONUtil.toJsonStr(item))));

        BulkRequest bulkRequest = new BulkRequest();
        for (ESMoviceInfo item : esMovieInfoList) {
            // RestHighLeverClient: 数据指定了索引库和类型就需要指明主键
            String id = UUID.randomUUID().toString(true);
            UpdateRequest updateRequest = new UpdateRequest();
            updateRequest.id(id);
            updateRequest.index(ElasticSearchConstants.INDEX_LIBRARY_NAME);
            updateRequest.type(ElasticSearchConstants.TABLE_NAME);
            updateRequest.doc(BeanUtil.beanToMap(item));
            // 表示如果数据不存在，那么就新增一条
            updateRequest.docAsUpsert(true);
            bulkRequest.add(updateRequest);
        }
        for (ESMoviceInfo item : esGreenMovieInfoList) {
            // RestHighLeverClient: 数据指定了索引库和类型就需要指明主键
            String id = UUID.randomUUID().toString(true);
            UpdateRequest updateRequest = new UpdateRequest();
            updateRequest.id(id);
            updateRequest.index(ElasticSearchConstants.GREEN_INDEX_LIBRARY_NAME);
            updateRequest.type(ElasticSearchConstants.TABLE_NAME);
            updateRequest.doc(BeanUtil.beanToMap(item));
            // 表示如果数据不存在，那么就新增一条
            updateRequest.docAsUpsert(true);
            bulkRequest.add(updateRequest);
        }

        client.bulk(bulkRequest, RequestOptions.DEFAULT);
        LOG.info("定时任务完成");
    }

}
